package com.lm.kafka.example;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.Random;
import java.util.concurrent.ExecutionException;

public class KafkaDataGenerator {

    public static void main(String[] args) {
        // Kafka 配置
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.2.0.230:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 可选：提高吞吐量配置
        props.put(ProducerConfig.LINGER_MS_CONFIG, 20);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32 * 1024); // 32KB
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

        // 创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        try {
            // 生成测试数据
            generateTestData(producer, "your-topic", 1000);
        } finally {
            producer.close();
        }
    }

    private static void generateTestData(KafkaProducer<String, String> producer,
                                         String topic, int messageCount) {
        Random random = new Random();
        String[] messageTypes = {"INFO", "WARN", "ERROR", "DEBUG"};
        String[] services = {"user-service", "order-service", "payment-service", "inventory-service"};

        for (int i = 0; i < messageCount; i++) {
            // 生成随机消息内容
            String messageType = messageTypes[random.nextInt(messageTypes.length)];
            String service = services[random.nextInt(services.length)];
            long timestamp = System.currentTimeMillis();

            String message = String.format("{\"type\":\"%s\",\"service\":\"%s\",\"message\":\"test_message_%d\",\"timestamp\":%d}",
                    messageType, service, i, timestamp);

            // 创建消息记录
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, null, message);

            // 发送消息（同步方式）
            try {
                RecordMetadata metadata = producer.send(record).get();
                System.out.printf("Sent message %d: %s (partition: %d, offset: %d)%n",
                        i, message, metadata.partition(), metadata.offset());

                // 控制发送速率
                Thread.sleep(100);

            } catch (InterruptedException | ExecutionException e) {
                System.err.println("Error sending message: " + e.getMessage());
            }
        }
    }
}